/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.cluster.impl.kryo;

import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

import org.sabayframework.cluster.ComputerId;
import org.sabayframework.cluster.NodeId;
import org.sabayframework.log.Log;

import com.esotericsoftware.kryonet.Client;
import com.esotericsoftware.kryonet.Connection;
import com.esotericsoftware.kryonet.Server;

/**
 * Class is responsible for bookkeeping of network connections
 * */
public class NetworkManager {
	
	private Server server;
	private final Map<ComputerId, Client> clientConnections;
	private final Map<ComputerId, Integer> serverConnections;
    private final KryoListener kryoListener;
    private final ComputerId localId;
	
	public NetworkManager(KryoListener kryoListener, Server server, ComputerId localId) {
		super();
		this.localId = localId;
		this.kryoListener = kryoListener;
		this.server = server;
		this.clientConnections = Collections.synchronizedMap(new HashMap<ComputerId, Client>());
		this.serverConnections = Collections.synchronizedMap(new HashMap<ComputerId, Integer>());
	}
	public void addConnection(ComputerId clientId, Integer connId){
		
		if(!serverConnections.containsKey(clientId) && !clientConnections.containsKey(clientId)){
			Log.debug("Connected connection ", connId);
			serverConnections.put(clientId, connId);
		}
	}
	@SuppressWarnings("unchecked")
	public void removeConnection(Integer connId){
		Map.Entry<ComputerId, Integer>[] entries = 
				serverConnections.entrySet().toArray(new Map.Entry[0]);
		
		for(Map.Entry<ComputerId, Integer> entry: entries){
			if(entry.getValue().equals(connId)){
				serverConnections.remove(entry.getKey());
				Log.debug("Disconnected connection ", connId);
				return;
			}
		}
		Map.Entry<ComputerId, Client>[] clients = 
				clientConnections.entrySet().toArray(new Map.Entry[0]);
		
		for(Map.Entry<ComputerId, Client> clientEntry : clients){
			final Client client = clientEntry.getValue();
			if(connId.intValue() == client.getID()){
				clientConnections.remove(clientEntry.getKey());
				Log.debug("Disconnected connection ", connId);
				return;
			}
		}
	}
	
	public void removeConnections(final NodeId id){
		Client client = clientConnections.remove(id);
		if(client != null) {
			client.stop();
		}
		Integer connectionId = serverConnections.remove(id);
		if(connectionId!=null){
			Connection[] conns = server.getConnections();
			for(int i=0;i<conns.length;++i)
				if(conns[i].getID() == connectionId.intValue()){
					conns[i].close();
					Log.debug("Closed connection ", connectionId);
					break;
				}
		}
		
	}
	public void sendData(ComputerId serverId, NetworkPacket data) throws IOException{
		data.setRemoteId(localId);
		if(localId.equals(serverId)){
			//skip network for local computer
			kryoListener.processNetworkPacket(data);
			return;
		}
		//try to find existing connection to server
		Integer connectionId = serverConnections.get(serverId);
		if(connectionId==null){
			Client client = clientConnections.get(serverId);
			if(client == null){
				client = new Client();
				KryoServer.initKryo(client.getKryo());
				client.start();
				client.addListener(kryoListener);
				client.connect(5000, serverId.getAddress(), serverId.getPort());
				clientConnections.put(serverId, client);
				Log.debug("New client ", client);
			}
			Log.debug("Sending using ",client,": ", data);
			client.sendTCP(data);
		} else{
			Log.debug("Sending using connection ",connectionId,": ", data);
			server.sendToTCP(connectionId.intValue(), data);
		}
	}

}
